package com.ruoyi.common.service.impl;

import cn.hutool.json.JSONObject;
import com.ruoyi.common.core.domain.entity.PtWsMessage;
import com.ruoyi.common.enums.WorkType;
import com.ruoyi.common.mapper.PtWsMessageMapper;
import com.ruoyi.common.service.WebSocket;
import com.ruoyi.common.utils.DateUtils;
import com.ruoyi.common.utils.SecurityUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
public class WebSocketImpl implements WebSocket {
    /**
     * 在线连接数（线程安全）
     */
    private final AtomicInteger connectionCount = new AtomicInteger(0);

    /**
     * 线程安全的无序集合（存储会话）
     */
    private final CopyOnWriteArraySet<WebSocketSession> sessions = new CopyOnWriteArraySet<>();

    @Autowired
    private PtWsMessageMapper ptWsMessageMapper;

    @Override
    public void handleOpen(WebSocketSession session) {
        sessions.add(session);
        int count = connectionCount.incrementAndGet();
        log.info("a new connection opened，current online count：{}", count);
    }

    @Override
    public void handleClose(WebSocketSession session) {
        sessions.remove(session);
        int count = connectionCount.decrementAndGet();
        log.info("a new connection closed，current online count：{}", count);
    }

    @Override
    public void handleMessage(WebSocketSession session, String message) {
        // 只处理前端传来的文本消息，并且直接丢弃了客户端传来的消息
        log.info("received a message：{}", message);
    }

    @Override
    public void sendMessage(WebSocketSession session, String message) {
        this.sendMessage(session, new TextMessage(message));
    }

    @Override
    public void sendMessage(String userId, TextMessage message) {
        Optional<WebSocketSession> userSession = sessions.stream().filter(session -> {
            if (!session.isOpen()) {
                return false;
            }
            Map<String, Object> attributes = session.getAttributes();
            if (!attributes.containsKey("uid")) {
                return false;
            }
            String uid = (String) attributes.get("uid");
            return uid.equals(userId);
        }).findFirst();
        if (userSession.isPresent()) {
            try {
                userSession.get().sendMessage(message);
            } catch (IOException e) {
                throw new RuntimeException("发送消息失败");
            }
        } else {
            // TODO 用户未登录，执行存库操作，再该用户登录时再次发送message
            PtWsMessage ptWsMessage = new PtWsMessage();
            ptWsMessage.setProductUser(SecurityUtils.getUsername());
            ptWsMessage.setProductTime(DateUtils.getNowDate());
            ptWsMessage.setStatus(Boolean.FALSE);
            ptWsMessage.setConsumer(userId);
            ptWsMessageMapper.insertPtWsMessage(ptWsMessage);
            List<PtWsMessage> list = getMessage(userId);
            System.out.println(list);
        }
    }

    @Override
    public void sendMessage(String userId, String message) {
        this.sendMessage(userId, new TextMessage(message));
    }

    @Override
    public void sendMessage(String userId, WorkType type, String id) {
        JSONObject message = new JSONObject(2);
        message.set("type", type.getTable()).set("id", id);
        this.sendMessage(userId, new TextMessage(message.toString()));
    }

    @Override
    public void sendMessage(WebSocketSession session, TextMessage message) {
        try {
            session.sendMessage(message);
        } catch (IOException e) {
            throw new RuntimeException("发送消息失败");
        }
    }

    @Override
    public void broadCast(String message) {
        for (WebSocketSession session : sessions) {
            if (!session.isOpen()) {
                continue;
            }
            this.sendMessage(session, message);
        }
    }

    @Override
    public void broadCast(TextMessage message) throws IOException {
        for (WebSocketSession session : sessions) {
            if (!session.isOpen()) {
                continue;
            }
            session.sendMessage(message);
        }
    }

    @Override
    public void handleError(WebSocketSession session, Throwable error) {
        log.error("websocket error：{}，session id：{}", error.getMessage(), session.getId());
        log.error("", error);
    }

    @Override
    public Set<WebSocketSession> getSessions() {
        return sessions;
    }

    @Override
    public int getConnectionCount() {
        return connectionCount.get();
    }

    @Override
    public List<PtWsMessage> getMessage(String user) {
        return ptWsMessageMapper.getMessagesByUser(user);
    }
}
